/*
 * Copyright (c) 2013, marco.tamburelli@gmail.com
 * All rights reserved.
 * 
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met: 
 * 
 * 1. Redistributions of source code must retain the above copyright notice, this
 *    list of conditions and the following disclaimer. 
 * 2. Redistributions in binary form must reproduce the above copyright notice,
 *    this list of conditions and the following disclaimer in the documentation
 *    and/or other materials provided with the distribution. 
 * 
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
 * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE LIABLE FOR
 * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
 * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
 * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
 * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */
package org.omorphdb.core.storage;

import static org.omorphdb.core.storage.StorageConstants.REFERENCED_DOC;
import static org.omorphdb.core.storage.StorageConstants.REMOVED_DOC;
import static org.omorphdb.core.storage.StorageConstants.ROOT_LEVEL_DOC;
import static org.omorphdb.core.value.ValueConstants._ARRAY;
import static org.omorphdb.core.value.ValueConstants._BOOLEAN;
import static org.omorphdb.core.value.ValueConstants._BYTE;
import static org.omorphdb.core.value.ValueConstants._CHAR;
import static org.omorphdb.core.value.ValueConstants._DATE;
import static org.omorphdb.core.value.ValueConstants._DOCUMENT;
import static org.omorphdb.core.value.ValueConstants._DOUBLE;
import static org.omorphdb.core.value.ValueConstants._FLOAT;
import static org.omorphdb.core.value.ValueConstants._INT;
import static org.omorphdb.core.value.ValueConstants._LONG;
import static org.omorphdb.core.value.ValueConstants._SHORT;
import static org.omorphdb.core.value.ValueConstants._STRING;

import java.io.IOException;

import org.omorphdb.core.exception.CorruptedStorageException;
import org.omorphdb.core.index.unique.DefaultIndex;
import org.omorphdb.core.io.Reader;
import org.omorphdb.core.io.Resource;
import org.omorphdb.core.io.ResourceFinder;
import org.omorphdb.core.storage.document.Document;
import org.omorphdb.core.storage.document.FieldFactory;
import org.omorphdb.core.value.ArrayValue;
import org.omorphdb.core.value.ObjectValue;
import org.omorphdb.core.value.Value;
import org.omorphdb.core.value.ValueFactory;

/**
 * Contains methods to read documents from the proper resource.
 */
class DocumentReader
{
	private final FieldFactory fieldFactory;
	private final ResourceFinder finder;
	private final Resource resource;

	private final DefaultIndex defaultIdx;
	private final Reader reader;

	/**
	 * Constructor
	 * 
	 * @param fieldFactory The specific {@link FieldFactory} implementation;
	 * @param finder The specific {@link ResourceFinder} implementation.
	 * 
	 * @throws IOException
	 */
	public DocumentReader(FieldFactory fieldFactory, ResourceFinder finder) throws IOException
	{
		this.fieldFactory = fieldFactory;
		this.finder = finder;

		this.resource = finder.acquireStorage();
		this.defaultIdx = new DefaultIndex(finder.acquireDefaultIndex());

		this.reader = this.resource.createReader();
	}

	/**
	 * Read the first root level document from the storage file.
	 * 
	 * @return
	 * @throws IOException
	 */
	public Document getFirst() throws IOException
	{
		this.reader.setCursor(0);

		return getNext();
	}

	/**
	 * This method should be called after at least one document has been read. <br>
	 * returns the next root level document if exists.
	 * 
	 * @return
	 * @throws IOException
	 */
	public Document getNext() throws IOException
	{
		if (this.reader.getCursor() >= this.resource.getResourceLimit()) return null;

		DocumentDescriptorImpl descr = readDescriptor();

		if (descr.status != DocumentStatus.ROOT_LEVEL)
		{
			this.reader.shift(descr.docSize);
			return getNext();
		}

		DocumentImpl doc = descr.createDocument(this.fieldFactory);

		for (int i = 0; i < descr.attributesCount; i++)
		{
			String attrName = this.reader.readString();
			doc.setNoUpdate(attrName, readValue());
		}

		return doc;
	}

	Document get(long pointer) throws IOException
	{
		this.reader.setCursor(pointer);

		DocumentDescriptorImpl descr = readDescriptor();

		if (descr.status == DocumentStatus.REMOVED)
			throw new CorruptedStorageException("Pointing to a removed document");

		DocumentImpl doc = new DocumentImpl(descr, this.fieldFactory);

		for (int i = 0; i < descr.attributesCount; i++)
		{
			String attrName = this.reader.readString();
			doc.setNoUpdate(attrName, readValue());
		}

		return doc;
	}

	private DocumentDescriptorImpl readDescriptor() throws IOException
	{
		long idx = this.reader.getCursor();

		char prefix = (char) this.reader.readByte();
		DocumentStatus status;

		switch (prefix)
		{
		case ROOT_LEVEL_DOC:
			status = DocumentStatus.ROOT_LEVEL;
			break;

		case REFERENCED_DOC:
			status = DocumentStatus.REFERENCED;
			break;

		case REMOVED_DOC:
			status = DocumentStatus.REMOVED;
			break;

		default:
			throw new CorruptedStorageException("Invalid format: type was expected  but found invalid symbol.");
		}

		long key = this.reader.readLong();

		int classID = this.reader.readInt();
		int attrCount = this.reader.readInt();
		int docLength = this.reader.readInt();

		return DocumentDescriptorImpl.define(status, classID, attrCount, idx, docLength, key);
	}

	private Value<?> readValue() throws IOException
	{
		byte type = this.reader.readType();

		return readValue(type);
	}

	private Value<?> readValue(byte type) throws IOException
	{
		switch (type)
		{
		case _BOOLEAN:
			return ValueFactory.createBoolean(this.reader.readBoolean());

		case _BYTE:
			return ValueFactory.createByte(this.reader.readByte());

		case _CHAR:
			return ValueFactory.createChar(this.reader.readChar());

		case _SHORT:
			return ValueFactory.createShort(this.reader.readShort());

		case _INT:
			return ValueFactory.createInteger(this.reader.readInt());

		case _LONG:
			return ValueFactory.createLong(this.reader.readLong());

		case _FLOAT:
			return ValueFactory.createFloat(this.reader.readFloat());

		case _DOUBLE:
			return ValueFactory.createDouble(this.reader.readDouble());

		case _STRING:
			byte[] bytes = this.reader.readStringAsBytes();
			return ValueFactory.createString(bytes);

		case _DATE:
			return ValueFactory.createDate(this.reader.readLong());

		case _DOCUMENT:
			long key = this.reader.readLong();

			DocumentReader parallelReader = new DocumentReader(this.fieldFactory, this.finder);
			Document doc = parallelReader.get(this.defaultIdx.search(key));

			return new ObjectValue(doc);

		case _ARRAY:
			int l = this.reader.readInt();
			Value<?>[] array = new Value<?>[l];

			for (int i = 0; i < l; i++)
				array[i] = readValue(this.reader.readByte());

			return new ArrayValue(array);

		default:
			throw new CorruptedStorageException("Invalid format: type was expected  but found invalid symbol.");
		}
	}
}
